// class defines reactor core for scala
package eventfax.core

import scala.actors._
import scala.collection.mutable.HashMap
import java.io.IOException
import java.nio._
import java.net._
import java.nio.channels.{SelectionKey,Selector,ServerSocketChannel,SocketChannel}
import java.nio.charset.{CharacterCodingException,Charset,CharsetDecoder}
import java.util.Iterator
import java.util.Set
import scala.collection.jcl.TreeMap
import java.util.Date

abstract class ReactorCore(starterBlock: ReactorCore => Unit) {
  val selector: Selector = Selector.open()
  var readBuffer: ByteBuffer = ByteBuffer.allocate(1024)
  var faxData: FaxData = new FaxData()
  var timerStore: TreeMap[Long,AbstractTimer] = new TreeMap[Long,AbstractTimer]
  var activeConnections: TreeMap[String,Connection] = new TreeMap[String,Connection]
  var bindingIndex = 100000

  starterBlock(this)
  start_reactor

  // start the reactor loop
  def start_reactor = {
    while(true) { 
      try {
        selector.select(50)
      } catch { case ex: IOException => println("IO Exception") }
      check_for_events()
      check_for_actor_messages
      check_for_timers()
    }
  }

  // check for socket events
  def check_for_events() {
    var keys: Set = selector.selectedKeys()
    var key_iterator = keys.iterator()
    while(key_iterator.hasNext()){
      val key: SelectionKey = (key_iterator.next()).asInstanceOf[SelectionKey]
      key_iterator.remove()
      if(key.isAcceptable()) accept_client(key)
      else if(key.isReadable()) read_data(key)
      else if(key.isWritable()) write_data(key)
      else if(key.isConnectable()) complete_external_connection(key)
    }
  }

  // checks incoming messages from other actors and delegates responses back to clients
  def check_for_actor_messages

  // method checks for timers which are ready to run
  def check_for_timers() {
    var now = new Date().getTime()
    var timer_ready = true
    while(!timerStore.isEmpty && timer_ready) {
      var k = timerStore.firstKey
      if(k > now) timer_ready = false
      else {
        var timer = timerStore.removeKey(k)
        timer match {
          case Some(one_shot_timer: Timer) => one_shot_timer.timer_method()
          case Some(periodic_timer: PeriodicTimer) => {
            periodic_timer.timer_method()
            add_periodic_timer(periodic_timer.interval)(periodic_timer.timer_method)
          }
          case None => println("Unmatched")
        }
      }
    }
  }

  // for external connections which were delayed, complete the connection now and invoke callback
  def complete_external_connection(key: SelectionKey) {
    var connection_channel: SocketChannel = (key.channel()).asInstanceOf[SocketChannel]
    var connection_instance: Connection = (key.attachment()).asInstanceOf[Connection]

    var conn_flag: Boolean = connection_channel.finishConnect()

    if(conn_flag) {
      var readKey: SelectionKey = connection_channel.register(selector,SelectionKey.OP_READ,connection_instance)
      val binding = create_binding
      connection_instance.decorate_handler(connection_channel,readKey,this,binding)
      addToActiveConnection(binding,connection_instance)
      connection_instance.connection_completed()
      // println("Connection is cmpleted")
      //key.cancel()
    }
  }

  // starts TCP Server
  def start_server(host: String,port: Int,connection_factory: ConnectionFactory) {
    var server: ServerSocketChannel = ServerSocketChannel.open()
    server.configureBlocking(false)
    server.socket().bind(new InetSocketAddress(host,port))
    server.register(selector,SelectionKey.OP_ACCEPT,connection_factory)
  }

  // invokes connect on specified IP and Port, and adds the socket to CONNECT event watchlist
  def connect(host: String,port: Int,connection_factory: ConnectionFactory): Connection = {
    var connection_channel: SocketChannel = SocketChannel.open()
    connection_channel.configureBlocking(false)
    var conn_flag: Boolean = connection_channel.connect(new InetSocketAddress(host,port))
    var connection_instance: Connection = connection_factory.create_connection()
    connection_instance.post_init()

    if(conn_flag) {
      var readKey: SelectionKey = connection_channel.register(selector,SelectionKey.OP_READ,connection_instance)
      val binding = create_binding
      connection_instance.decorate_handler(connection_channel,readKey,this,binding)
      addToActiveConnection(binding,connection_instance)
      connection_instance.connection_completed()
    } else {
      connection_channel.register(selector,SelectionKey.OP_CONNECT,connection_instance)
    }
    return connection_instance
  }

  // accepts a new client and invokes post_init and connection_completed
  def accept_client(key: SelectionKey): Unit = {
    var server: ServerSocketChannel = (key.channel()).asInstanceOf[ServerSocketChannel]
    var connection_factory = key.attachment().asInstanceOf[ConnectionFactory]

    var server_instance: Connection = connection_factory.create_connection()

    // if(connection_factory.connection_block != null) {
    //   connection_factory.connection_block(server_instance)
    // }
    var client: SocketChannel = server.accept();
    client.configureBlocking(false);
    var readKey: SelectionKey = client.register(selector, SelectionKey.OP_READ,server_instance);
    val binding = create_binding
    server_instance.decorate_handler(client,readKey,this,binding)
    addToActiveConnection(binding,server_instance)
    server_instance.post_init()
    return server_instance
  }



  // will read data and invoke appropriate callback on client
  def read_data(key: SelectionKey) {
    var client: SocketChannel = (key.channel()).asInstanceOf[SocketChannel]
    readBuffer.clear()

    var nothing_to_read = false
    var client_disconnected = false
    var data_length = 0

    while(!nothing_to_read && !client_disconnected) {
      try {
        data_length = client.read(readBuffer)
      } catch { case ex: IOException => client_disconnected = true }
      if(data_length == 0) nothing_to_read = true
      else if(data_length == -1) client_disconnected = true
      
      readBuffer.flip()
      faxData.append(readBuffer);
      readBuffer.clear()
    }
    faxData.decode()

    var client_instance: Connection = key.attachment().asInstanceOf[Connection]
    if(client_disconnected) {
      client_instance.close_connection()
      // client_instance.unbind
      // key.cancel()
      // client.close
    } else {
      client_instance.receive_data(faxData)
      faxData.clear()
    }
  }

  // write data to the specified connection
  def write_data(key: SelectionKey) {
    var client_instance: Connection = key.attachment().asInstanceOf[Connection]
    client_instance.write_and_schedule()
  }

  def add_timer(delay: Int)(timer_method: () => Unit) {
    var t_timer = Timer(delay,timer_method)
    timerStore.put(time_after(delay),t_timer)
  }
  
  def time_after(seconds: Int): Long = {
    var now = new Date().getTime()
    now + (seconds * 1000)
  }

  def add_periodic_timer(interval: Int)(timer_method: () => Unit) {
    var periodic_timer = PeriodicTimer(interval,timer_method)
    timerStore.put(time_after(interval),periodic_timer)
  }


  def create_binding: String = {
    bindingIndex += 1
    "BND_" + bindingIndex
  }

  def addToActiveConnection(binding: String,connection: Connection) = {
    activeConnections.put(binding,connection)
  }

}


